package com.atguigu.flink.demo04;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * @author admin
 * @date 2021/8/10
 */
public class RedisSink {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 从网络中读取数据
         * 参数：ip,段口，分隔符，重试次数
         */
        SocketTextStreamFunction socketText = new SocketTextStreamFunction("hadoop102", 9999, "\n", 3);
        DataStreamSource<String> source = env.addSource(socketText);

        // redis 连接配置
        FlinkJedisPoolConfig builder = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop102")
                .setPort(6379)
                .build();

        // redis 命令
        RedisMapper<String> redisMapper = new RedisMapper<String>() {

            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.HSET,"key_sss");
            }

            @Override
            public String getKeyFromData(String data) {
                return data+"_ByKey";
            }

            @Override
            public String getValueFromData(String data) {
                return data;
            }
        };


        org.apache.flink.streaming.connectors.redis.RedisSink<String> stringRedisSink =
                new org.apache.flink.streaming.connectors.redis.RedisSink<>(builder,redisMapper);

        source.addSink(stringRedisSink);


        env.execute();

    }

}
